package com.deep.flink.hudi;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimestampType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.types.Row;

import java.util.Arrays;


public class SyncCustomerRelationship {


    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);


        String inputTableSQL = "create table tbl_customer_relationship_mysql (\n" +
                "  id string PRIMARY KEY NOT ENFORCED,\n" +
                "  create_date_time string,\n" +
                "  update_date_time string,\n" +
                "  deleted string,\n" +
                "  customer_id string,\n" +
                "  first_id string,\n" +
                "  belonger string,\n" +
                "  belonger_name string,\n" +
                "  initial_belonger string,\n" +
                "  distribution_handler string,\n" +
                "  business_scrm_department_id string,\n" +
                "  last_visit_time string,\n" +
                "  next_visit_time string,\n" +
                "  origin_type string,\n" +
                "  itcast_school_id string,\n" +
                "  itcast_subject_id string,\n" +
                "  intention_study_type string,\n" +
                "  anticipat_signup_date string,\n" +
                "  `level` string,\n" +
                "  creator string,\n" +
                "  current_creator string,\n" +
                "  creator_name string,\n" +
                "  origin_channel string,\n" +
                "  `comment` string,\n" +
                "  first_customer_clue_id string,\n" +
                "  last_customer_clue_id string,\n" +
                "  process_state string,\n" +
                "  process_time string,\n" +
                "  payment_state string,\n" +
                "  payment_time string,\n" +
                "  signup_state string,\n" +
                "  signup_time string,\n" +
                "  notice_state string,\n" +
                "  notice_time string,\n" +
                "  lock_state string,\n" +
                "  lock_time string,\n" +
                "  itcast_clazz_id string,\n" +
                "  itcast_clazz_time string,\n" +
                "  payment_url string,\n" +
                "  payment_url_time string,\n" +
                "  ems_student_id string,\n" +
                "  delete_reason string,\n" +
                "  deleter string,\n" +
                "  deleter_name string,\n" +
                "  delete_time string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  delete_comment string,\n" +
                "  close_state string,\n" +
                "  close_time string,\n" +
                "  appeal_id string,\n" +
                "  tenant string,\n" +
                "  total_fee string,\n" +
                "  belonged string,\n" +
                "  belonged_time string,\n" +
                "  belonger_time string,\n" +
                "  transfer string,\n" +
                "  transfer_time string,\n" +
                "  follow_type string,\n" +
                "  transfer_bxg_oa_account string,\n" +
                "  transfer_bxg_belonger_name string\n" +
                ")WITH(\n" +
                "  'connector' = 'mysql-cdc',\n" +
                "  'hostname' = 'localhost',\n" +
                "  'port' = '3320',\n" +
                "  'username' = 'root',\n" +
                "  'password' = '123456',\n" +
                "  'server-time-zone' = 'Asia/Shanghai',\n" +
                "  'debezium.snapshot.mode' = 'initial',\n" +
                "  'database-name' = 'itcast_nev',\n" +
                "  'table-name' = 'customer_relationship'\n" +
                ") ";

        tableEnv.executeSql(inputTableSQL);


        String outputTableSQL = "create table edu_customer_relationship_hudi(\n" +
                "  id string PRIMARY KEY NOT ENFORCED,\n" +
                "  create_date_time string,\n" +
                "  update_date_time string,\n" +
                "  deleted string,\n" +
                "  customer_id string,\n" +
                "  first_id string,\n" +
                "  belonger string,\n" +
                "  belonger_name string,\n" +
                "  initial_belonger string,\n" +
                "  distribution_handler string,\n" +
                "  business_scrm_department_id string,\n" +
                "  last_visit_time string,\n" +
                "  next_visit_time string,\n" +
                "  origin_type string,\n" +
                "  itcast_school_id string,\n" +
                "  itcast_subject_id string,\n" +
                "  intention_study_type string,\n" +
                "  anticipat_signup_date string,\n" +
                "  `level` string,\n" +
                "  creator string,\n" +
                "  current_creator string,\n" +
                "  creator_name string,\n" +
                "  origin_channel string,\n" +
                "  `comment` string,\n" +
                "  first_customer_clue_id string,\n" +
                "  last_customer_clue_id string,\n" +
                "  process_state string,\n" +
                "  process_time string,\n" +
                "  payment_state string,\n" +
                "  payment_time string,\n" +
                "  signup_state string,\n" +
                "  signup_time string,\n" +
                "  notice_state string,\n" +
                "  notice_time string,\n" +
                "  lock_state string,\n" +
                "  lock_time string,\n" +
                "  itcast_clazz_id string,\n" +
                "  itcast_clazz_time string,\n" +
                "  payment_url string,\n" +
                "  payment_url_time string,\n" +
                "  ems_student_id string,\n" +
                "  delete_reason string,\n" +
                "  deleter string,\n" +
                "  deleter_name string,\n" +
                "  delete_time string,\n" +
                "  course_id string,\n" +
                "  course_name string,\n" +
                "  delete_comment string,\n" +
                "  close_state string,\n" +
                "  close_time string,\n" +
                "  appeal_id string,\n" +
                "  tenant string,\n" +
                "  total_fee string,\n" +
                "  belonged string,\n" +
                "  belonged_time string,\n" +
                "  belonger_time string,\n" +
                "  transfer string,\n" +
                "  transfer_time string,\n" +
                "  follow_type string,\n" +
                "  transfer_bxg_oa_account string,\n" +
                "  transfer_bxg_belonger_name string,\n" +
                "  part STRING\n" +
                ")\n" +
                "PARTITIONED BY (part)\n" +
                "WITH(\n" +
                "  'connector'='hudi',\n" +
                "  'path'= 'hdfs://localhost:9000/user/deeprado/hudi-warehouse/edu_customer_hudi', \n" +
                "  'table.type'= 'MERGE_ON_READ',\n" +
                "  'hoodie.datasource.write.recordkey.field'= 'id', \n" +
                "  'write.precombine.field'= 'create_date_time',\n" +
                "  'write.tasks'= '1',\n" +
                "  'write.rate.limit'= '2000', \n" +
                "  'compaction.tasks'= '1', \n" +
                "  'compaction.async.enabled'= 'true',\n" +
                "  'compaction.trigger.strategy'= 'num_commits',\n" +
                "  'compaction.delta_commits'= '1',\n" +
                "  'changelog.enabled'= 'true'\n" +
                ") ";

        tableEnv.executeSql(outputTableSQL);

        String syncSQL= "insert into edu_customer_hudi \n" +
                "select *, CAST(CURRENT_DATE AS STRING) AS part from tbl_customer_mysql ";

        tableEnv.executeSql(syncSQL);

         env.execute("Api_Sink");
    }
}
